package leo.peersim.paxos;

import java.io.File;
import java.util.Vector;

import leo.peersim.common.IFSAutomaton;
import leo.peersim.common.LamportEventComparator;
import leo.peersim.common.LogicalClock;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import peersim.config.Configuration;
import peersim.core.CommonState;
import peersim.core.Linkable;
import peersim.core.Node;
import peersim.edsim.EDProtocol;
import peersim.edsim.EDSimulator;
import peersim.transport.Transport;

/**
 * <p>Paxos consensus protocol used for implementing a fault-tolerant
 * Finite-State Automaton (FSA)</p>
 * 
 * @author leo
 * @version June 19, 2009
 */
public class Paxos implements EDProtocol, IFSAutomaton {
	
	private Logger logger = Logger.getLogger(this.getClass().getName());
	
	private final int _transportProtocolID;
	private final int _groupManagementProtocolID;
	
	private long _timestamp;
	
	// latest ballot took part in phase 1
	private LogicalClock _latestPhase1Ballot;
	
	// latest ballot accepted a value in phase 2
	private LogicalClock _latestPhase2Ballot;
	
	// latest accepted value in phase 2
	private long _latestAcceptedValue;
	
	private Proposer _proposer;
	private Acceptor _acceptor;
	private Learner _learner;
	
	protected Node _estimateOfRecentLeader;
	
	private Vector<ReplicaCommand> _executedCommands;

	@Deprecated
	protected long _timeout = 5*140;
	
	/**
	 * Constructor called one-time by the simulator engine for initializing protocol
	 * from configuration file
	 * 
	 * @param prefix
	 */
	public Paxos(String prefix) {
		this._transportProtocolID = Configuration.getPid(prefix + "." + "transport");
		this._groupManagementProtocolID = Configuration.getPid(prefix + "." + "collegamenti");
		
		String log4jcfg = Configuration.getString(prefix + "." + "log4jcfg");
		PropertyConfigurator.configure((new File(new File("cfg"), log4jcfg)).getAbsolutePath());
	}
	
	/**
	 * Initialize state local to each node
	 */
	private void init() {
		this._timestamp = 0;
		
		this._latestPhase1Ballot = new LogicalClock(CommonState.getNode(), this._timestamp);
		this._latestPhase2Ballot = new LogicalClock(CommonState.getNode(), this._timestamp);
		this._latestAcceptedValue = 0;
		
		this._proposer = new Proposer(this);
		this._acceptor = new Acceptor(this);
		this._learner = new Learner(this);
		
		this._executedCommands = new Vector<ReplicaCommand>();
		
		this._estimateOfRecentLeader = CommonState.getNode();
	}
	
	/**
	 * Called by the simulator engine to put the protocol on each node of the network
	 */
	public Paxos clone() {
		Paxos p = null;
		try {
			p = (Paxos) super.clone();
			logger.trace("cloned!");
		} catch (CloneNotSupportedException e) {
			logger.fatal("Error on cloning!"); // It should never happens
		}
		
		p.init();
		
		return p;
	}

	/**
	 * Dispatcher for incoming messages
	 */
	@Override
	public void processEvent(Node hostingNode, int protocolID, Object event) {
		if (event instanceof ClientRequest) {
			this.requestOperation((ClientRequest)event);
		} else if (event instanceof Prepare) {
			this.processEvent((Prepare)event);
		} else if (event instanceof Promise) {
			this.processEvent((Promise)event);
		} else if (event instanceof Refuse) {
			this.processEvent((Refuse)event);
		} else if (event instanceof Reject) {
			this.processEvent((Reject)event);
		} else if (event instanceof Decide) {
			this.processEvent((Decide)event);
		} else if (event instanceof Accept) {
			this.processEvent((Accept)event);
		} else if (event instanceof Timeout) {
			this._proposer.timeout((Timeout) event);			
		} else {
			logger.fatal("Dispatching error!");
		}
		
	}
	
	private void processEvent(Prepare m) {
		//forwarding to Acceptor
		this._acceptor.manage(m);
	}
	
	private void processEvent(Promise m) {
		//forwarding to Leader
		this._proposer.manage(m);
	}
	
	private void processEvent(Refuse m) {
		//forwarding to Leader
		this._proposer.manage(m);
	}
	
	private void processEvent(Accept m) {
		//forwarding to acceptor
		this._acceptor.manage(m);
	}
	
	private void processEvent(Reject m) {
		//forwarding to acceptor
		this._proposer.manage(m);
	}
	
	private void processEvent(Decide m) {
		//forwarding to learner
		
		ReplicaCommand r = this._learner.manage(m);
		
		//forward to the leader only if the decide comes from the leader's node
		//FIXME tricky
		if(m.getBallotId().getLamportNodeID()==CommonState.getNode().getID()) {
			if(m.getSenderNode()==CommonState.getNode()) {
				this._proposer.decide(m.getBallotId(), r);
			}
		}
	}
	
	/**
	 * Handler for request message sent by a client
	 * 
	 * @param operation
	 */
	private void requestOperation(ClientRequest operation) {
		
		//FIXME cross-cutting concern: does make sense to replicate 
		//ClientRequest to on each node?
		//ClientRequest cloned = operation.clone();
		
		this._proposer.submit(operation);
	}
	
	
	protected void enqueueCommand(String command) {
		//this.commandQueue.add(command);
	}
	
	@Override
	public Vector<String> getCommandHistory() {
		Vector<String> commandHistory = new Vector<String>(); 
		for (ReplicaCommand r: this._executedCommands) {
			commandHistory.add(r.getCommandName());
		}
		return commandHistory;
	}
	
	@Override
	public Vector<IClientRequest> getClientRequests() {
		return this._proposer.getAllClientRequests();
	}

	/**
	 * 
	 * @return the Lamport Logical Clock of the hosting node
	 */
	protected LogicalClock getLogicalClock() {
		return new LogicalClock(CommonState.getNode(), this._timestamp);
	}

	/**
	 * 
	 * @return next ballot to be issued to the group for a new Paxos round
	 */
	protected LogicalClock getNewBallot() {
		long latestBallotTimestamp = this._latestPhase1Ballot.getLamportTimestamp();
		this._latestPhase1Ballot = new LogicalClock(CommonState.getNode(), (latestBallotTimestamp+1));
		return this._latestPhase1Ballot;
	}
	
	protected boolean updateLatestBallot(LogicalClock candidateBallot) {
		boolean res = this.checkLatestPhaseABallot(candidateBallot);
		if(true) {
			this._latestPhase1Ballot = candidateBallot;
			//logger.debug(CommonState.getNode().getID() + "latest ballot updated: _" + this._latestBallot + " candidate:" + candidateBallot);
		}
		return res;
	}
	
	/**
	 * 
	 * @see LamportEventComparator
	 * 
	 * @param candidateBallot
	 * @return true if {@code candidateBallot} is recent
	 */
	protected boolean checkLatestPhaseABallot(LogicalClock candidateBallot) {
		LamportEventComparator l = new LamportEventComparator();
		int res = l.compare(candidateBallot, this._latestPhase1Ballot );
		if(res>=0) {
			return true;
		} else {
			return false;
		}
	}
	
	protected LogicalClock getLatestPhase1Ballot() {
		return this._latestPhase1Ballot;
	}
	
	protected LogicalClock getLastPhase2Ballot() {
		return this._latestPhase2Ballot;
	}
	
	protected long getLatestAcceptValue() {
		return this._latestAcceptedValue;
	}
	
	protected void updateLatestAcceptValue(long proposal) {
		this._latestAcceptedValue = proposal;
	}
	
	protected void updateLatestPhase2Ballot(LogicalClock ballotId) {
		this._latestPhase2Ballot = ballotId;
	}

	/**
	 * 
	 * @param message
	 */
	protected void broadcastToGroup(PaxosMessage message) {
		//send to itself
		Node localNode = CommonState.getNode();
		EDSimulator.add(0, message, localNode, CommonState.getPid());

		Linkable l = (Linkable)localNode.getProtocol( this._groupManagementProtocolID  );
		for (int i = 0; i<l.degree(); i++) {
			Node recipient = l.getNeighbor(i);
			
			Transport trLayer = (Transport) recipient.getProtocol(_transportProtocolID);		
			trLayer.send(localNode, recipient, message, CommonState.getPid());
		}	
	}

	/**
	 * 
	 * @param message
	 * @param recipient
	 */
	protected void send(PaxosMessage message, Node recipient) {
		Node localNode = CommonState.getNode();
		
		if(localNode==recipient) {
			EDSimulator.add(0, message, localNode, CommonState.getPid());
		} else {
			Transport trLayer = (Transport) recipient.getProtocol(_transportProtocolID);
			trLayer.send(localNode, recipient, message, CommonState.getPid());
		}
	}

	protected void executeCommand(ReplicaCommand rc) {
		rc.execute();
		this._executedCommands.add(rc);
	}

	public Vector<ReplicaCommand> getExecutedCommandHistory() {
		return this._executedCommands;
	}
	
}
